package com.google.code.stalkernet.async.factory;

import java.io.EOFException;
import java.nio.ByteBuffer;
import java.util.concurrent.TimeoutException;

import com.google.code.stalkernet.async.io.AsyncSocketChannel;
import com.google.code.stalkernet.node.NodeID;
import com.google.code.stalkernet.node.NodeUtils;
import com.google.code.stalkernet.peer.AsyncSocketChannelPeer;
import com.google.code.stalkernet.reactor.Reactor;
import com.google.code.stalkernet.reactor.Reactor.ScheduledTaskHandle;

public class AsyncSocketChannelPeerFactory
    implements AsyncFactory <AsyncSocketChannel, AsyncSocketChannelPeer>
{
    private final NodeID nodeID;
    private final Reactor reactor;

    public AsyncSocketChannelPeerFactory (NodeID nodeID, Reactor reactor)
    {
        if (nodeID == null)
            throw new IllegalArgumentException ("Node ID is null");

        if (reactor == null)
            throw new IllegalArgumentException ("Reactor is null");

        this.nodeID = nodeID;
        this.reactor = reactor;
    }

    @Override
    public Handle produce (
        AsyncSocketChannel source,
        Callback <? super AsyncSocketChannelPeer> callback)
    {
        if (source == null)
            throw new IllegalArgumentException ("Source is null");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        AsyncSocketChannelWrapper asyncSocketChannelWrapper =
            new AsyncSocketChannelWrapper (source, callback);

        return asyncSocketChannelWrapper.start ();
    }

    private class AsyncSocketChannelWrapper implements Handle
    {
        private final AsyncSocketChannel asyncSocketChannel;
        private final Callback <? super AsyncSocketChannelPeer> callback;

        private Handle readHandle = null;
        private Handle writeHandle = null;
        private ScheduledTaskHandle timeoutHandle = null;

        private NodeID remoteNodeID = null;
        private boolean nodeIDSent = false;
        private boolean done = false;

        public AsyncSocketChannelWrapper (
            AsyncSocketChannel asyncSocketChannel,
            Callback <? super AsyncSocketChannelPeer> callback)
        {
            if (asyncSocketChannel == null)
                throw new IllegalArgumentException (
                    "Async socket channel is null");

            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.asyncSocketChannel = asyncSocketChannel;
            this.callback = callback;
        }

        public Handle start ()
        {
            receiveNodeIDLength ();
            sendNodeID ();
            setTimeout ();

            return done ? null : this;
        }

        @Override
        public boolean cancel ()
        {
            if (done) return false;
            else
            {
                if (readHandle != null)
                    readHandle.cancel ();

                if (writeHandle != null)
                    writeHandle.cancel ();

                if (timeoutHandle != null)
                    timeoutHandle.cancel ();

                done = true;
                return true;
            }
        }

        private void receiveNodeIDLength ()
        {
            final ByteBuffer byteBuffer = ByteBuffer.allocate (4);
            readHandle = asyncSocketChannel.read (
                byteBuffer,
                new Callback <Integer> ()
                {
                    @Override
                    public void onSuccess (Integer result)
                    {
                        if (result == null)
                            throw new IllegalArgumentException (
                                "Result is null");

                        readHandle = null;

                        int l = result.intValue ();

                        if (l < 0)
                            error (new EOFException ());
                        else if (byteBuffer.remaining () > 0)
                            readHandle = asyncSocketChannel.read (
                                byteBuffer, this);
                        else
                        {
                            byteBuffer.flip ();
                            receiveNodeID (byteBuffer.getInt ());
                        }
                    }

                    @Override
                    public void onError (Throwable error)
                    {
                        if (error == null)
                            throw new IllegalArgumentException (
                                "Error is null");

                        readHandle = null;

                        error (error);
                    }
                });
        }

        private void receiveNodeID (int length)
        {
            if (length < 0)
                throw new IllegalArgumentException ("Length < 0");

            final ByteBuffer byteBuffer = ByteBuffer.allocate (length);
            readHandle = asyncSocketChannel.read (
                byteBuffer,
                new Callback <Integer> ()
                {
                    @Override
                    public void onSuccess (Integer result)
                    {
                        if (result == null)
                            throw new IllegalArgumentException (
                                "Result is null");

                        readHandle = null;

                        int l = result.intValue ();

                        if (l < 0)
                            error (new EOFException ());
                        else if (byteBuffer.remaining () > 0)
                            readHandle = asyncSocketChannel.read (
                                byteBuffer, this);
                        else
                        {
                            byteBuffer.flip ();
                            byte [] nodeIDBytes =
                                new byte [byteBuffer.remaining ()];
                            byteBuffer.get (nodeIDBytes);
                            nodeIDReceived (
                                NodeUtils.createNodeID (nodeIDBytes));
                        }
                    }

                    @Override
                    public void onError (Throwable error)
                    {
                        if (error == null)
                            throw new IllegalArgumentException (
                                "Error is null");

                        readHandle = null;

                        error (error);
                    }
                });
        }

        private void sendNodeID ()
        {
            byte [] nodeIDBytes = nodeID.getBytes ();
            int nodeIDBytesCount = nodeIDBytes.length;
            final ByteBuffer byteBuffer = ByteBuffer.allocate (
                nodeIDBytesCount + 4);
            byteBuffer.putInt (nodeIDBytesCount);
            byteBuffer.put (nodeIDBytes);
            byteBuffer.flip ();

            writeHandle = asyncSocketChannel.write (
                byteBuffer,
                new Callback <Integer>()
                {
                    @Override
                    public void onSuccess (Integer result)
                    {
                        if (result == null)
                            throw new IllegalArgumentException (
                                "Result is null");

                        writeHandle = null;

                        if (byteBuffer.remaining () > 0)
                            writeHandle =
                                asyncSocketChannel.write (byteBuffer, this);
                        else nodeIDSent ();
                    }

                    @Override
                    public void onError (Throwable error)
                    {
                        if (error == null)
                            throw new IllegalArgumentException (
                                "Error is null");

                        writeHandle = null;

                        error (error);
                    }
                });
        }

        private void setTimeout ()
        {
            timeoutHandle = reactor.schedule (
                new Runnable () {
                    @Override
                    public void run ()
                    {
                        timeoutHandle = null;

                        error (new TimeoutException ());
                    }
                },
                System.currentTimeMillis () + 1000);
        }

        private void nodeIDReceived (NodeID remoteNodeID)
        {
            this.remoteNodeID = remoteNodeID;

            checkSuccess ();
        }

        private void nodeIDSent ()
        {
            nodeIDSent = true;

            checkSuccess ();
        }

        private void checkSuccess ()
        {
            if (remoteNodeID != null && nodeIDSent)
            {
                timeoutHandle.cancel ();

                done = true;
                callback.onSuccess (
                    new AsyncSocketChannelPeer (
                        remoteNodeID, asyncSocketChannel));
            }
        }

        private void error (Throwable error)
        {
            cancel ();
            callback.onError (error);
        }
    }
}
